package com.min.main.kafka;

import cn.hutool.log.Log;
import com.min.kafka.entity.KMessage;
import com.min.kafka.topic.KafkaTopic;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Optional;

@Component
public class ConsumerListener {

    private final Log log = Log.get();

    @KafkaListener(topics = KafkaTopic.BATCH_DEMO, containerFactory = "batchFactory", errorHandler = "kafkaListenerErrorHandler")
    public void batchConsumerListener(List<ConsumerRecord<?, KMessage>> records) {

        if (records == null) {
            log.error("Topic[{}] Receive Error", KafkaTopic.BATCH_DEMO);
        } else {
            for (ConsumerRecord<?, KMessage> record : records) {
                final Object key = record.key();
                Optional<KMessage> kafkaMessage = Optional.ofNullable(record.value());
                if (kafkaMessage.isPresent()) {
                    log.debug("Topic[{}] receive: {}", KafkaTopic.BATCH_DEMO, record.value());
                } else {
                    log.error("Topic[{}] receive null ", KafkaTopic.BATCH_DEMO);
                }
            }
        }
    }

    @KafkaListener(topics = KafkaTopic.SINGLE_DEMO, containerFactory = "baseFactory", errorHandler = "kafkaListenerErrorHandler")
    public void singleConsumerListener(ConsumerRecord<?, KMessage> record) {

        if (record == null) {
            log.error("Topic[{}] Receive Error", KafkaTopic.SINGLE_DEMO);
        } else {
            Optional<KMessage> kafkaMessage = Optional.ofNullable(record.value());
            if (kafkaMessage.isPresent()) {
                log.debug("Topic[{}] receive: {}", KafkaTopic.SINGLE_DEMO, record.value());
            } else {
                log.error("Topic[{}] receive null ", KafkaTopic.SINGLE_DEMO);
            }
        }
    }

}
